package ldh.im;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import ldh.im.dto.base.Dto;
import ldh.im.dto.base.ErrorResp;
import ldh.im.exception.ImServiceException;
import ldh.im.handle.*;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.function.Consumer;

public class ClientChannelService {

//    private static final Logger LOGGER = LoggerFactory.getLogger(ClientChannelService.class);
    private ExecutorService writeCallBackExecutor = Executors.newFixedThreadPool(1);  // 回调处理器
    private ExecutorService serverMessageExecutor = Executors.newFixedThreadPool(1);  // 服务器返回的消息处理器

    private static final int MAX_LENGTH = 1024 * 20;
    private int warnSize = 1000;

    private Channel channel = null;
    private Bootstrap clientBootstrap;
    private NioEventLoopGroup group;
    private String host;
    private int port;

    private Consumer<Dto> serverMessageConsumer;

    private ConcurrentHashMap<String, Object> session = new ConcurrentHashMap<>();  // channelId对应的回话
    private ConcurrentHashMap<String, Consumer<? extends Dto>> callbackMap = new ConcurrentHashMap<>(); // 回调函数

    public ClientChannelService(String address, int port, Consumer<Dto> consumer) {
        if (consumer == null) {
            throw new IllegalArgumentException("consumer不能为空");
        }
        this.host = address;
        this.port = port;
        this.serverMessageConsumer = wrapServerThreadConsumer(consumer);

        buildClient();
    }

    public synchronized void connection() throws ImServiceException, InterruptedException {
        if (channel == null || !channel.isActive() || !channel.isOpen()) {
            //发起异步连接请求，绑定连接端口和host信息
            final ChannelFuture future = clientBootstrap.connect(host, port).sync();
            future.awaitUninterruptibly();
            if (!future.isSuccess()) {
                throw new ImServiceException("连接服务器失败");
            }
            channel = future.channel();
        }
    }

    public boolean isCallbackEnd() {
        return callbackMap.size() == 0;
    }

    public void close() {
        int size = callbackMap.size();
        if (size > 0) {
//            LOGGER.warn("回调列表中还有数据没处理完, 未处理数量： {}", size);
        }
        if (isActive()) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
        writeCallBackExecutor.shutdown();
        serverMessageExecutor.shutdown();
    }

    public ChannelFuture closeFuture() {
        return channel.closeFuture();
    }

    public <T extends Dto>Future<T> write(Dto dto) throws InterruptedException, TimeoutException, ImServiceException {
        if (!isActive()) {
            connection();
        }
        String seq = dto.getLocalSeq();
        if (callbackMap.containsKey(seq)) {
            throw new ImServiceException("localSeq重复!!!, seq=" + seq);
        }
        ImFuture<T> future = new ImFuture<>();
        Consumer<T> consumer = buildCallback(future);
        callbackMap.put(seq, consumer);
        int size = callbackMap.size();
        if (size >= warnSize) {
//            LOGGER.warn("发送数据过快，请减缓发送数据速度!!, size: {}", size);
        }
        channel.writeAndFlush(dto);
        return future;
    }

    public <T extends Dto>T write(Dto dto, long timeout, TimeUnit timeUnit) throws InterruptedException, ImServiceException, TimeoutException {
        if (!isActive()) {
            connection();
        }
        String seq = dto.getLocalSeq();
        if (callbackMap.containsKey(seq)) {
            throw new ImServiceException("localSeq重复!!!, seq=" + seq);
        }
        ImFuture<T> future = new ImFuture<>();
        Consumer<T> consumer = buildCallback(future);
        callbackMap.put(seq, consumer);
        int size = callbackMap.size();
        if (size >= warnSize) {
//            LOGGER.warn("发送数据过快，请减缓发送数据速度!!, size: {}", size);
        }
        long nanos = timeUnit.toNanos(timeout);
        channel.writeAndFlush(dto);
        return future.get(timeout, timeUnit);
    }

    public void write(Dto dto, Consumer<Dto> consumer) throws ImServiceException, InterruptedException {
        if (!isActive()) {
            connection();
        }
        String seq = dto.getLocalSeq();
        if (callbackMap.containsKey(seq)) {
            throw new ImServiceException("localSeq重复!!!, seq=" + seq);
        }
        callbackMap.put(seq, wrapThreadConsumer(consumer));
        int size = callbackMap.size();
        if (size >= warnSize) {
//            LOGGER.warn("发送数据过快，请减缓发送数据速度!!, 解压数量：{}", size);
        }
        channel.writeAndFlush(dto);
    }

    public void writeAsync(Dto dto) throws InterruptedException, ImServiceException {
        if (!isActive()) {
            connection();
        }
        channel.writeAndFlush(dto);
    }

    public boolean isActive() {
        return channel != null && channel.isActive();
    }

    public void setChannel(Channel channel2) {
        if (channel == null || !channel.isOpen()) {
            this.channel = channel2;
            return;
        }
        throw new RuntimeException("channel已经设置，不需要再次设置");
    }

    public void put(String key, Object value) {
        session.put(key, value);
    }

    public void setAuthToken(String authToken, int authCode) {
        session.put(ImConfig.AUTH_TOKEN, authToken);
        session.put(ImConfig.AUTH_CODE, authCode);
    }

    public boolean isLogin() {
        return session.containsKey(ImConfig.AUTH_TOKEN);
    }

    public int getAuthCode() {
        if (session.containsKey(ImConfig.AUTH_CODE)) {
            return (int) session.get(ImConfig.AUTH_CODE);
        }
        throw new RuntimeException("未设置AUTH_TOKEN");
    }

    public Object get(String key) {
        return session.get(key);
    }

    public void cleanSession() {
        session.clear();
    }

    private <T extends Dto>Consumer<T> buildCallback(ImFuture<T> future) {
        Consumer<T> consumer = to->{
            if (to instanceof ErrorResp) {
                ErrorResp errorRec = (ErrorResp) to;
                future.setThrowable(new ImServiceException(errorRec.getErrorCode() + ":" + errorRec.getErrorMsg()));
            }
            future.setSuccess(to);
        };
        return consumer;
    }

    private Consumer<Dto> wrapThreadConsumer(Consumer<Dto> toConsumer) {
        return to->{
            if (toConsumer != null) {
                writeCallBackExecutor.execute(()->toConsumer.accept(to));
            }
        };
    }

    private Consumer<Dto> wrapServerThreadConsumer(Consumer<Dto> toConsumer) {
        return to->{
            serverMessageExecutor.execute(()->toConsumer.accept(to));
        };
    }

    private void buildClient()  {
        group = new NioEventLoopGroup(1);
        clientBootstrap = new Bootstrap();
        clientBootstrap.group(group)
                .channel(NioSocketChannel.class)  // 使用NioSocketChannel来作为连接用的channel类
                .handler(new ChannelInitializer<SocketChannel>() { // 绑定连接初始化器
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
//                        LOGGER.info("正在连接中...");
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline
                                .addLast(new ExceptionOutHandler())
                                .addLast(new ConnectionHandler())
                                .addLast(new LengthFieldBasedFrameDecoder(MAX_LENGTH, ImConfig.CLIENT_HEAD_FIELD_LENGTH, 4))
                                .addLast(new ImCheckHandler())
                                .addLast(new TransferDataCoderHandler(ClientChannelService.this))
                                .addLast(new CToCoderHandler(ClientChannelService.this.callbackMap, ClientChannelService.this.serverMessageConsumer))
                                .addLast(new ExceptionInHandler())
                        ;

                    }
                });
    }


}
